package com.gitee.xmhzzz.emqx.bridge.rocketmq.listener;


import com.gitee.xmhzzz.emqx.client.listener.IMqttListener;
import com.gitee.xmhzzz.emqx.client.model.message.IMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Random;

/**
 * @ClassName MqttBridgeRocketmqListener
 * @Description
 * @Author wzq
 * @Date 2024/1/22 16:51
 * @Version 1.0
 */
@Slf4j
@Component
public class MqttBridgeRocketmqListener implements IMqttListener {

    @Autowired(required = false)
    private RocketMQTemplate rocketMQTemplate;

    private String topic;

    public MqttBridgeRocketmqListener() {
        this.topic = "#";
    }

    @Override
    public String getTopic() {
        return this.topic;
    }

    @Override
    public String getType() {
        return null;
    }

    @Override
    public void onMessage(IMessage message) {
        Random rand = new Random();
        asyncSendMq("device-topic","device",String.valueOf(rand.nextInt(100)),message.getMessageContent());
        log.info("==> MqttBridgeRocketmqListener topic[{}]  message[{}]",message.getHeaders().get("topic"),message.getMessageContent());
    }

    public void asyncSendMq(String topic,String tag,String key,String massage){
        Message<String> springMsg = MessageBuilder.withPayload(massage).setHeader(RocketMQHeaders.KEYS, key).build();
        rocketMQTemplate.asyncSend(topic +":"+tag, springMsg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 发送成功逻辑
                log.info("asyncSend ok, messageId: {}" , sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                // 发送失败逻辑
                log.error("asyncSend failed, payload: {}", massage);
            }
        },2000);
    }
}
